package com.boot.study;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StreamWordCountJava {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 从文件中读取数据
        String inputPath = "D:/WorkSpace/idea/Flink/src/main/resources/hello.txt";
        DataStreamSource<String> ds = env.readTextFile(inputPath);
        SingleOutputStreamOperator<WordCount> dataStream = ds.flatMap(new FlatMapFunction<String, WordCount>() {
            public void flatMap(String value, Collector<WordCount> out) {
                if (StringUtils.isNotEmpty(value)) {
                    String[] arr = value.split(" ");
                    for (String word : arr) {
                        out.collect(new WordCount(word, 1));
                    }
                }
            }
        }).keyBy("word")
//                .sum("count")
                .reduce(new ReduceFunction<WordCount>() {
                    public WordCount reduce(WordCount value1, WordCount value2) {
                        return new WordCount(value1.word, value1.count + value2.count);
                    }
                });

        dataStream.print("wc").setParallelism(1);
        env.execute("stream wc java");
    }

    public static class WordCount {
        public String word;
        public Integer count;

        public WordCount() {
        }

        public WordCount(String word, Integer count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return "{word:" + this.word + ",count:" + this.count + "}";
        }
    }
}
